package com.lovecaa.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 消费者
 * 创建时间： 2023年10月20日 23:32:00
 *
 * @author Mr.Tao
 * @since 1.0.0
 */
public class OrderConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultMQPushConsumer");
        consumer.setNamesrvAddr("192.168.192.110:9876");
        // 订阅消费主题，并且获取所有Tag的消息
        consumer.subscribe("OrderTest", "*");
        // 跟并发消费的监听器不同的是这里采用的是一个顺序消费的监听器
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext context) {
                for (int i = 0; i < list.size(); i++) {
                    System.out.println(i + "消息消费成功:" + new String(list.get(i).getBody(), StandardCharsets.UTF_8));
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("消费者启动成功...");
    }
}
